package cn.doitedu.dw.sql

import cn.doitedu.commons.util.{SparkUtil, YieeDateUtils}
import org.apache.spark.sql.internal.StaticSQLConf.CATALOG_IMPLEMENTATION
import org.apache.spark.sql.{Dataset, Row, SaveMode, SparkSession}

import scala.collection.immutable

/**
 * 活跃用户留存分析报表计算
 * 目标表： ADS_APL_ART_REC  活跃用户留存记录表
 * CREATE TABLE ADS_APL_ART_REC(
 * dt         string ,  -- 计算日期
 * act_dt     string ,  -- 活跃日期
 * rt_days    int    ,  -- 留存天数
 * rt_users   int    ,  -- 留存人数
 * )
 * stored as parquet;
 *
 * 源表：  DWS_APL_UCA_RNG   用户连续访问区间记录表
 *
 */
object ADS_APL_ART_REC {
  def main(args: Array[String]): Unit = {
    /*val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local")
      .enableHiveSupport()  //开启对hive的支持
      .getOrCreate()*/
    val spark = SparkUtil.getSparkSession(this.getClass.getSimpleName,confMap = Map(CATALOG_IMPLEMENTATION.key->"hive"))

    import spark.implicits._

    // 读取  源表：  用户连续访问区间记录表
    val rng: Dataset[Row] = spark.read.table("doit13.dws_apl_uca_rng").where("dt='2020-03-14' and rng_end='9999-12-31'")
    rng.show(10,false)

    sys.exit()


    // 展开区间
    val actDt = rng.flatMap(row=>{
      val guid = row.getAs[Long]("guid")
      val rng_start = row.getAs[String]("rng_start")
      // 比如，今天是3-14号， 而 rng_start = 3-10号,那么，就要展开成：
      /***
       * 3-10
       * 3-11
       * 3-12
       * 3-13
       */
      // 先求  当日-rng_start 的差值
      import cn.doitedu.commons.util.YieeDateUtils._
      val diff: Long = dateDiff(rng_start, "2020-03-14")

      for(i<- 0 until diff.toInt) yield (guid,dateAdd(rng_start,i.toInt))
    }).toDF("guid","act_dt")

    // 汇总聚合
    actDt.createTempView("act")
    val res = spark.sql(
      """
        |select
        |'2020-03-14' as dt,
        |act_dt,
        |datediff('2020-03-14',act_dt) as rt_days,
        |count(1) as rt_users
        |from act
        |group by act_dt
        |""".stripMargin)

    res.write.mode(SaveMode.Append).saveAsTable("doit13.ADS_APL_ART_REC")

    spark.close()
  }

}
